// -*-mode:c++; coding:utf-8-*-

#include <iostream>
#include <sstream>
#include <string>

using namespace std;

#include <boost/cast.hpp>
#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>

#include <xbase_utility.hpp>
#include <naming_strategy.hpp>

#include "io_service_pool_impl.hpp"
#include "client_environment.hpp"

using namespace xbase;

typedef boost::system::error_code error_code;

typedef ::kb::network_byteorder byteorder;
typedef ::kb::binary_iarchive<byteorder> iarchive;

client_environment *the_env = NULL;

boost::mutex quit_mutex;


std::string dump_value(const std::string &value, const std::string &column) {
	if(column == "info:tabletInfo")
	{
		protocol::tablet_info t;
		protocol::iarchive in(value.data(), value.size());
		return (t.load(in) 
			? t.to_string()
			: string("decode failed, data = ") + dump_string_hex(value));
	}
	else if(column == "info:serverAddress")
	{
		protocol::server_address t;
		protocol::iarchive in(value.data(), value.size());
		return (t.load(in)
			? t.to_string()
			: string("decode failed, data = ") + dump_string_hex(value));
	}
	else if(column == "info:serverStartCode")
	{
		uint64_t t;
		protocol::iarchive in(value.data(), value.size());
		return (in.load(&t)
			? boost::lexical_cast<std::string>(t) + " (" + dump_timestamp(t) + ')'
			: std::string("decode failed, data = ") + dump_string_hex(value));
	}
	else if((column == "info:splitA") || (column == "info:splitB"))
	{
		std::string t;
		protocol::iarchive in(value.data(), value.size());
		return (in.load(&t)
			? dump_string_char(t)
			: string("decode failed, data = ") + dump_string_hex(value));
	}
	else			// logs:
	{
		return dump_string_char(value);
	}
}

void on_get_next_row(tablet_scanner::ptr_type ts_ptr, row_result::ptr_type row_ptr, const error_code &ec)
{
	if(ec)
	{
		if(ec != xbase::error::eos)
		{
			std::cout << "get next row failed : " << ec
				  << " - " << ec.message() << std::endl;
		}
		quit_mutex.unlock();
	}
	else
	{
		cout << dump_string_char(row_ptr->m_row_name) << std::endl;

		row_result::cell_vector::iterator iter = row_ptr->m_cell_vector.begin();
		while (iter != row_ptr->m_cell_vector.end())
		{
			cell &c = *iter;
			unsigned version_count = c.get_version_count();

			std::cout << "\t" << c.m_column_name
				  << " versions(" << version_count << ")" << std::endl;

			for (unsigned i=0;i<version_count;i++)
			{
				cell::version v;
				c.get_version(i, v);
				cout << "\t\t" << v.first << " (" << dump_timestamp(v.first) << ")\t"
				     << (v.second.empty()
					 ? std::string("#<deletion record>")
					 : dump_value(v.second, c.m_column_name)) << std::endl;
			}

			++iter;
		}

		ts_ptr->async_get_next_row(boost::bind(on_get_next_row, ts_ptr, _1, _2));
	}
}

int main(int argc, char **argv)
{
	LOG_CONFIG("../../conf/log4cplus.tablet_server.conf");

	if(argc != 3)
	{
		std::cout << "test_dump_meta_tablet <master ip> <master port>" << std::endl;
		return 0;
	}
	
	//
	// 首先初始化service_pool并启动
	// 

	boost::shared_ptr<io_service_pool> pool(new impl::single_io_service());
	client_options_ptr options(new client_options);
	client_environment env(options, pool, argv[1], ::atoi(argv[2]));
	the_env = &env;
	quit_mutex.lock();

	tablet_scan_info si(".METADATA_1ST_LEVEL", "", "");
	si.m_column_names.clear();
	si.m_column_names.push_back("*");
	si.m_latest_time_stamp = 0;
	si.m_oldest_time_stamp = 0;
	si.m_ttl = 0;
	si.m_version_number = 0;
	si.m_options = 1;
	si.m_num_row = 100;

	basic_sync_handler<> sh;
	env.get_root_meta_cache().async_get_root_meta(sh.make_handler());
	sh.wait();
	if(sh.get_error_code())
	{
		std::cout << "got root meta address failed: " << sh.get_error_code().message();
		return -1;
	}
	LOG(DEBUG, "got root meta tablet info = " << env.get_root_meta_cache().get_root_tablet_info().to_string());
	LOG(DEBUG, "got root meta server address = " << env.get_root_meta_cache().get_root_server_address().to_string());
	const boost::asio::ip::tcp::endpoint ep = env.get_root_meta_cache().get_root_server_address().to_endpoint();

	tablet_scanner::ptr_type ts_ptr(new tablet_scanner(env.get_io_service_pool(),
							   ep,
							   env.get_session_manager(), si));

	ts_ptr->async_get_next_row(boost::bind(on_get_next_row, ts_ptr, _1, _2));

	quit_mutex.lock();
	quit_mutex.unlock();

	return 0;
}
